package com.hoveagle.kafka.listener;

import com.hoveagle.kafka.handler.StatusMessageHandler;
import com.hoveagle.kafka.consumer.KafkaConfig;
import com.hoveagle.kafka.consumer.MessageConsumer;
import com.hoveagle.kafka.consumer.MessageConsumerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * define a kafka message listener which includes some message consumers, you can start multiple consumers at the same time,
 * kafka resources can be released gracefully when system shut down.
 * @author hoveagle
 */
public class KafkaMessageListener {

    List<MessageConsumer> messageConsumers;

    public KafkaMessageListener(){
        messageConsumers = new ArrayList<>();
        messageConsumers.add(MessageConsumerFactory.newInstance(KafkaConfig.CONFIG1, new StatusMessageHandler()));
    }

    public void start(){
        registerHook();
        messageConsumers.forEach(MessageConsumer::start);
    }

    private void registerHook(){
        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run(){
                messageConsumers.forEach(MessageConsumer::stop);
            }
        });
    }
}
